package com.guaoran.study.odps;

import lombok.extern.slf4j.Slf4j;
import ru.yandex.clickhouse.ClickHouseConnection;
import ru.yandex.clickhouse.ClickHouseDataSource;
import ru.yandex.clickhouse.ClickHouseDatabaseMetadata;
import ru.yandex.clickhouse.response.ClickHouseResultBuilder;
import ru.yandex.clickhouse.settings.ClickHouseProperties;

import java.sql.*;
import java.util.*;

/**
 * @author : gu cheng
 * @description : click house demo
 * @date : 2021/5/6 14:19
 */
@Slf4j
public class ClickHouseDemo {
    private static String address = "jdbc:clickhouse://172.26.20.120:8123/DEMO";
    private static String username = "u1";
    private static String password = "123456";
    private static String databaseName = "DEMO";
    private static Integer socketTimeout = 600000;

    public static Connection getConn() {
        ClickHouseConnection conn = null;
        ClickHouseProperties properties = new ClickHouseProperties();
        properties.setUser(username);
        properties.setPassword(password);
        properties.setDatabase(databaseName);
        properties.setSocketTimeout(socketTimeout);
        ClickHouseDataSource clickHouseDataSource = new ClickHouseDataSource(address, properties);
        try {
            conn = clickHouseDataSource.getConnection();
            return conn;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return null;
    }


    /**
     * 获取表信息，因为 jdbc 的实现没有获取到表备注，这里直接重写
     *
     * @param schemaPattern
     * @param tableNamePattern
     * @param types
     * @param connection
     * @see ClickHouseDatabaseMetadata#getTables(String, String, String, String[])
     * @return
     * @throws SQLException
     */
    private ResultSet getTables(String schemaPattern, String tableNamePattern, String[] types, ClickHouseConnection connection) throws SQLException {
        // 重写了
        String sql = "select database, name, engine,comment from system.tables where 1 = 1";
        if (schemaPattern != null) {
            sql += " and database like '" + schemaPattern + "'";
        }
        if (tableNamePattern != null) {
            sql += " and name like '" + tableNamePattern + "'";
        }
        sql += " order by database, name";

        // 重写了
        Statement statement = connection.createStatement();
        ResultSet result = statement.executeQuery(sql);

        ClickHouseResultBuilder builder = ClickHouseResultBuilder
                .builder(10)
                .names(
                        "TABLE_CAT",
                        "TABLE_SCHEM",
                        "TABLE_NAME",
                        "TABLE_TYPE",
                        "REMARKS",
                        "TYPE_CAT",
                        "TYPE_SCHEM",
                        "TYPE_NAME",
                        "SELF_REFERENCING_COL_NAME",
                        "REF_GENERATION")
                .types(
                        "String",
                        "String",
                        "String",
                        "String",
                        "String",
                        "String",
                        "String",
                        "String",
                        "String",
                        "String");

        List<String> typeList = types != null ? Arrays.asList(types) : null;
        while (result.next()) {
            List<String> row = new ArrayList<>();
            row.add(result.getString(1));
            row.add(result.getString(1));
            row.add(result.getString(2));
            String type, e = result.getString(3).intern();
            switch (e) {
                case "View":
                case "MaterializedView":
                case "Merge":
                case "Distributed":
                case "Null":
                    type = "VIEW"; // some kind of view
                    break;
                case "Set":
                case "Join":
                case "Buffer":
                    type = "OTHER"; // not a real table
                    break;
                default:
                    type = "TABLE";
                    break;
            }
            row.add(type);

            // 重写了
            row.add(result.getString(4));
            for (int i = 4; i < 9; i++) {
                row.add(null);
            }
            if (typeList == null || typeList.contains(type)) {
                builder.addRow(row);
            }
        }
        result.close();
        return builder.build();
    }

    public static ResultSet getTables(String catalog, String schemaPattern, String tableNamePattern, String[] types,Connection connection) throws SQLException {
        String sql = "select " +
                "database, name, engine,comment " +
                "from system.tables " +
                "where 1 = 1";
        if (schemaPattern != null) {
            sql += " and database like '" + schemaPattern + "'";
        }
        if (tableNamePattern != null) {
            sql += " and name like '" + tableNamePattern + "'";
        }
        sql += " order by database, name";
        Statement statement = connection.createStatement();
        ResultSet result = statement.executeQuery(sql);

        ClickHouseResultBuilder builder = ClickHouseResultBuilder.builder(10);
        builder.names("TABLE_CAT", "TABLE_SCHEM", "TABLE_NAME", "TABLE_TYPE", "REMARKS", "TYPE_CAT", "TYPE_SCHEM", "TYPE_NAME", "SELF_REFERENCING_COL_NAME", "REF_GENERATION");
        builder.types("String", "String", "String", "String", "String", "String", "String", "String", "String", "String");

        List typeList = types != null ? Arrays.asList(types) : null;
        while (result.next()) {
            List<String> row = new ArrayList<>();
            row.add(result.getString(1));
            row.add(result.getString(1));
            row.add(result.getString(2));
            String type, e = result.getString(3).intern();
            if (e == "View" || e == "MaterializedView" || e == "Merge" || e == "Distributed" || e == "Null") {
                type = "VIEW"; // some kind of view
            } else if (e == "Set" || e == "Join" || e == "Buffer") {
                type = "OTHER"; // not a real table
            } else {
                type = "TABLE";
            }
            row.add(type);
            row.add(result.getString(4));
            for (int i = 4; i < 9; i++) {
                row.add(null);
            }
            if (typeList == null || typeList.contains(type)) {
                builder.addRow(row);
            }
        }
        result.close();
        return builder.build();
    }

    public static void test(String databaseName, Connection connection) throws Exception{
        // 获取表信息
        ResultSet resultSet = getTables(null, databaseName, null, new String[] { "TABLE" },connection);
        ResultSetMetaData rsmd = resultSet.getMetaData();
        List<Map<String, String>> list = new ArrayList<>();
        while (resultSet.next()) {
            String tableName = resultSet.getString("TABLE_NAME");
            String dataSourceName = resultSet.getString("TABLE_SCHEM");
            String desc = resultSet.getString("REMARKS");

            System.out.println(dataSourceName+"="+tableName+"="+desc);

            ResultSet rs = connection.getMetaData().getColumns(null, dataSourceName, tableName, "%");
            while (rs.next()) {
                String name = rs.getString("COLUMN_NAME");
                String comment = rs.getString("REMARKS");
                String type = rs.getString("TYPE_NAME");
                System.out.println("dataSourceName="+dataSourceName+"\ttableName="+tableName+"\ttableComment="+desc +"\tcolumnName="+name+"\tcomment="+comment+"\ttype="+type);
            }

            Map<String, String> row = new HashMap<>();
            for (int i = 1; i <= rsmd.getColumnCount(); i++) {
                row.put(rsmd.getColumnName(i), resultSet.getString(rsmd.getColumnName(i)));
            }
            list.add(row);
        }
        while (resultSet.next()) {
            String tableName = resultSet.getString("TABLE_SCHEM");
            String dataSourceName = resultSet.getString("TABLE_NAME");
            String desc = resultSet.getString("REMARKS");

            System.out.println(dataSourceName+"="+tableName+"="+desc);

        }
    }

    public static List<Map<String, String>> exeSql(Connection connection,String sql) {
        log.info("cliockhouse 执行sql：" + sql);
        try {
            Statement statement = connection.createStatement();
            ResultSet results = statement.executeQuery(sql);
            ResultSetMetaData rsmd = results.getMetaData();
            List<Map<String, String>> list = new ArrayList<>();
            while (results.next()) {
                Map<String, String> row = new HashMap<>();
                for (int i = 1; i <= rsmd.getColumnCount(); i++) {
                    row.put(rsmd.getColumnName(i), results.getString(rsmd.getColumnName(i)));
                }
                list.add(row);
            }
            return list;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return null;
    }

    public static void main(String[] args) throws Exception {
        Connection connection = getConn();
        test("demo",connection);

//        //查询数据库
//        String sqlDB = "show databases";
//        //查看表
//        String sqlTab = "use demo ;show tables";
//        //查询ontime数据量
//        String sqlCount = "select *  from hits_v1 limit 10";
//        List<Map<String, String>> maps = exeSql(connection, sqlDB);
//        System.out.println(maps);
//        List<Map<String, String>> maps1 = exeSql(connection, sqlTab);
//        System.out.println(maps1);
//        List<Map<String, String>> maps2 = exeSql(connection, sqlCount);
//        System.out.println(maps2);
    }
}
